package kvraft

import (
	"6.824/labgob"
	"6.824/labrpc"
	"6.824/raft"
	"6.824/utils"
	"bytes"
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

const Debug = false

func DPrintf(format string, a ...interface{}) (n int, err error) {
	if Debug {
		log.Printf(format, a...)
	}
	return
}

type Op struct {
	// Your definitions here.
	// Field names must start with capital letters,
	// otherwise RPC will break.
	Key     string
	Value   string
	Pattern int64
	Op      OpType
}
type OpType int

type ChanMsg struct {
	value string
	term  int
	err   Err
}

const (
	PUT     OpType = 0
	APPEND  OpType = 1
	GET     OpType = 2
	CONFIRM OpType = 3
)

type KVServer struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg
	dead    int32 // set by Kill()

	maxraftstate int // snapshot if log grows this big
	persister    *raft.Persister

	data          map[string]string
	patternSet    map[int64]bool
	lstApplyIndex int
	reqChannels   map[int]chan<- ChanMsg
	// Your definitions here.
}

func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	// Your code here.
	op := Op{Key: args.Key, Op: GET}
	v, err := kv.Start(op)
	reply.Err = err
	if err == OK {
		reply.Value = v
	}
	//log.Printf("server reply.err:%v\n", reply.Err)
}
func (kv *KVServer) Confirm(args *ConfirmArgs, reply *ConfirmReply) {
	// Your code here.
	op := Op{Pattern: args.Pattern, Op: CONFIRM}
	_, err := kv.Start(op)
	reply.Err = err

	//log.Printf("server reply.err:%v\n", reply.Err)
}

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	// Your code here.
	var op Op

	if args.Op == "Put" {

		op = Op{args.Key, args.Value, args.Pattern, PUT}
	} else {
		op = Op{args.Key, args.Value, args.Pattern, APPEND}
	}

	_, err := kv.Start(op)
	reply.Err = err
}

func (kv *KVServer) Start(op Op) (Value string, Err Err) {
	i, t, l := kv.rf.Start(op)
	if !l {
		Err = ErrWrongLeader
	} else {
		kv.mu.Lock()
		if c, ok := kv.reqChannels[i]; ok {
			close(c)
		}
		c := make(chan ChanMsg)
		kv.reqChannels[i] = c
		kv.mu.Unlock()
		select {
		case x, ok := <-c:
			if ok {

				if x.term == t {
					Err = x.err
					Value = x.value
				} else {
					Err = ErrFailToCommit
				}
				close(c)

			} else {

				Err = ErrFailToCommit

			}

		case <-time.After(time.Millisecond * reqTimeout):
			Err = ErrFailToCommit
		}

	}
	return
}

//
// the tester calls Kill() when a KVServer instance won't
// be needed again. for your convenience, we supply
// code to set rf.dead (without needing a lock),
// and a killed() method to test rf.dead in
// long-running loops. you can also add your own
// code to Kill(). you're not required to do anything
// about this, but it may be convenient (for example)
// to suppress debug output from a Kill()ed instance.
//
func (kv *KVServer) Kill() {
	atomic.StoreInt32(&kv.dead, 1)
	kv.rf.Kill()
	kv.mu.Lock()
	defer kv.mu.Unlock()
	for _, v := range kv.reqChannels {
		close(v)
	}
	kv.reqChannels = make(map[int]chan<- ChanMsg)
	// Your code here, if desired.
}

func (kv *KVServer) killed() bool {
	z := atomic.LoadInt32(&kv.dead)
	return z == 1
}

func (kv *KVServer) RecvFromRaft() {
	que := make(chan raft.ApplyMsg, 10)
	go func() {
		for kv.killed() == false {

			msg, ok := <-que

			//log.Printf("me:%v commit index:%v", kv.me, msg.CommandIndex)
			if ok {
				if msg.CommandValid {
					kv.ApplyOp(msg.Command.(Op), msg.CommandIndex, msg.CommandTerm)
				} else if msg.SnapshotValid {
					kv.recvSnapshot(msg.Snapshot, msg.SnapshotIndex)
				} else {
					panic("wrong apply")
				}
			}
		}
	}()

	for kv.killed() == false {
		msg, ok := <-kv.applyCh
		if ok {
			que <- msg
		}

	}
}
func (kv *KVServer) ApplyOp(op Op, index int, term int) {
	kv.mu.Lock()
	defer kv.mu.Unlock()

	//log.Printf("try send to channel:%v", index)
	c, ok := kv.reqChannels[index]
	var msg ChanMsg
	kv.lstApplyIndex = index

	switch op.Op {
	case PUT:
		if _, ok := kv.patternSet[op.Pattern]; !ok {

			if op.Value == "" {
				if _, ok := kv.data[op.Key]; ok {
					delete(kv.data, op.Key)
				}
			} else {
				kv.data[op.Key] = op.Value
			}
			kv.patternSet[op.Pattern] = true

		}
		//kv.Print("INFO", "pattern add:%v", op.Pattern)
		msg = ChanMsg{err: OK, term: term}

	case APPEND:
		if _, ok := kv.patternSet[op.Pattern]; !ok {
			if op.Value != "" {
				if _, ok := kv.data[op.Key]; ok {
					//kv.data[op.Key] = strings.Join([]string{kv.data[op.Key], op.Value}, "")
					kv.data[op.Key] += op.Value
				} else {
					kv.data[op.Key] = op.Value
				}
			}
			kv.patternSet[op.Pattern] = true

			/*if kv.me == 1 {

				kv.Print("INFO", "append k:%v v:%v", op.Key, op.Value)
			}*/
		}
		//log.Printf("threads:%v", runtime.NumGoroutine())
		//log.Printf("set size:%v chan size:%v", len(kv.patternSet), len(kv.reqChannels))
		//kv.Print("INFO", "pattern add:%v", op.Pattern)
		msg = ChanMsg{err: OK, term: term}
	case GET:
		if v, ok := kv.data[op.Key]; ok {
			msg = ChanMsg{v, term, OK}
		} else {
			msg = ChanMsg{err: ErrNoKey, term: term}
		}
	case CONFIRM:
		//kv.Print("INFO", "try delete pattern:%v", op.Pattern)
		if _, ok := kv.patternSet[op.Pattern]; ok {
			delete(kv.patternSet, op.Pattern)
		}
		msg = ChanMsg{err: OK, term: term}
	default:
		panic(fmt.Sprintf("wrong op type:%v", op.Op))
	}
	if ok {
		delete(kv.reqChannels, index)
		go func() {
			c <- msg
		}()

	}

	if kv.maxraftstate != -1 && kv.persister.RaftStateSize() >= kv.maxraftstate {
		kv.doSnapshot()
	}
}
func (kv *KVServer) sizeWatcherLoop() {
	kv.mu.Lock()
	maxsiz := kv.maxraftstate
	persister := kv.persister
	kv.mu.Unlock()
	for kv.killed() == false {
		time.Sleep(time.Millisecond * sizeWatcherSleepTime)
		if persister.RaftStateSize() > maxsiz*9/10 {
			kv.mu.Lock()
			kv.doSnapshot()
			kv.mu.Unlock()
		}

	}
}
func (kv *KVServer) doSnapshot() {

	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	if e.Encode(kv.data) != nil || e.Encode(kv.patternSet) != nil {
		panic("fail to encode")
	}
	kv.rf.Snapshot(kv.lstApplyIndex, w.Bytes())

}
func (kv *KVServer) recvSnapshot(snapshot []byte, index int) {
	kv.mu.Lock()
	defer kv.mu.Unlock()
	e := labgob.NewDecoder(bytes.NewBuffer(snapshot))
	if e.Decode(&kv.data) != nil || e.Decode(&kv.patternSet) != nil {
		panic("fail to decode")
	}
	utils.Assert(index >= kv.lstApplyIndex)
	kv.lstApplyIndex = index
}

//
// servers[] contains the ports of the set of
// servers that will cooperate via Raft to
// form the fault-tolerant key/value service.
// me is the index of the current server in servers[].
// the k/v server should store snapshots through the underlying Raft
// implementation, which should call persister.SaveStateAndSnapshot() to
// atomically save the Raft state along with the snapshot.
// the k/v server should snapshot when Raft's saved state exceeds maxraftstate bytes,
// in order to allow Raft to garbage-collect its log. if maxraftstate is -1,
// you don't need to snapshot.
// StartKVServer() must return quickly, so it should start goroutines
// for any long-running work.
//
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate

	kv.data = make(map[string]string)
	kv.patternSet = make(map[int64]bool)
	kv.reqChannels = make(map[int]chan<- ChanMsg)
	// You may need initialization code here.

	kv.lstApplyIndex = 0
	kv.applyCh = make(chan raft.ApplyMsg)
	kv.persister = persister
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)

	// You may need initialization code here.
	go kv.RecvFromRaft()
	if kv.maxraftstate != -1 {
		//go kv.sizeWatcherLoop()
	}

	return kv
}
